package com.com.zs21cp.watermark;


import com.com.zs21cp.ToolBean.ClickSource;
import com.com.zs21cp.ToolBean.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;


/**
 * @program: flink
 * @description:
 * @author: Arctic
 * @create: 2024-03-12 22:39
 */
public class WindowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStreamSource<Event> source =
//                (DataStreamSource<Event>) env.fromElements(
//                        new Event("Marry","./home",100L),
//                        new Event("alice","./home",100L),
//                        new Event("Mark","./home",100L),
//                        new Event("John","./home",100L)
//                );
//        KeyedStream<Event, String> EventStringKeyedStream = source.keyBy(data -> data.user);
//        //滚动时间时间窗口
//        EventStringKeyedStream
//                .countWindow(10,2)//滑动窗口的计数器
//                ;

        DataStreamSource<Event> source = env.addSource(new ClickSource());
        SingleOutputStreamOperator<Event> source1 = source.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));


        SingleOutputStreamOperator<Tuple2<String, Long>> map = source1.map(new MapFunction<Event, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Event value) throws Exception {
                return Tuple2.of(value.user, 1L);
            }
        });
        KeyedStream<Tuple2<String, Long>, String> keyby = map.keyBy(data -> data.f0);
        WindowedStream<Tuple2<String, Long>, String, TimeWindow> window = keyby.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        window.reduce(new ReduceFunction<Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                return Tuple2.of(value1.f0,value1.f1+value2.f1);
            }
        }).print();
        env.execute();
    }
}